package com.hudson.broker.handler.process;

import com.hudson.broker.handler.Process;
import com.hudson.broker.monitor.BizMonitor;
import com.hudson.broker.utils.StoreUtil;
import com.hudson.codec.domain.Message;
import com.hudson.codec.domain.RemotingCommand;
import com.hudson.common.enums.MessageTypeEnum;
import com.hudson.core.cache.CacheBuilder;
import com.hudson.core.cache.HudsonCache;
import io.netty.channel.ChannelHandlerContext;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.nio.charset.StandardCharsets;

public class DefaultProduceProcesser implements Process {
    private static final Logger logger = LoggerFactory.getLogger(DefaultProduceProcesser.class);

    HudsonCache<Long, String> MESSAGE_CACHE = CacheBuilder.<Long, String>newBuilder()
            .maximumSize(10000)
            .build();

    @Override
    public RemotingCommand process(ChannelHandlerContext context, RemotingCommand remotingCommand) {
        if (MESSAGE_CACHE.contains(remotingCommand.getOpaque())) {
            logger.info("message:{}重试成功", remotingCommand.getOpaque());
            return RemotingCommand.builder()
                    .type(MessageTypeEnum.SEND_MESSAGE_RESP.getType())
                    .opaque(remotingCommand.getOpaque())
                    .body(MESSAGE_CACHE.getIfPresent(remotingCommand.getOpaque()).getBytes(StandardCharsets.UTF_8))
                    .build();
        }

        Message message = Message.deserialize(remotingCommand.getBody());


        String msgId = StoreUtil.putMessage(message);

        MESSAGE_CACHE.put(remotingCommand.getOpaque(), msgId);

        BizMonitor.PRODUCE_SPEED.getAndIncrement();

        return RemotingCommand.builder()
                .type(MessageTypeEnum.SEND_MESSAGE_RESP.getType())
                .opaque(remotingCommand.getOpaque())
                .body(msgId.getBytes(StandardCharsets.UTF_8))
                .build();
    }
}
